package org.smartboot.sosa.rpc.process;

import java.util.concurrent.Executors;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.smartboot.sosa.rpc.RpcConfig;
import org.smartboot.sosa.rpc.RpcUnit;
import org.smartboot.sosa.rpc.Session;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.EventTranslatorTwoArg;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

/**
 * 基于Disruptor实现的业务层协议消息处理器
 *
 * @author Seer
 * @version DisruptorProcessor.java, v 0.1 2015年11月24日 下午2:04:11 Seer Exp.
 */
public class DisruptorProcessor<T> implements MessageProcessor<T> {
	private static final Logger LOGGER = LogManager.getLogger(DisruptorProcessor.class);
	private RingBuffer<RpcUnit<T>> ringBuffer;
	private Disruptor<RpcUnit<T>> disruptor;
	private RpcConfig rpcConfig;

	private EventTranslatorTwoArg<RpcUnit<T>, Session, T> TRANSLATOR = new EventTranslatorTwoArg<RpcUnit<T>, Session, T>() {
		@Override
		public void translateTo(RpcUnit<T> event, long sequence, Session session, T t) {
			event.setSession(session);
			event.setUnit(t);
		}
	};

	@SuppressWarnings("unchecked")
	@Override
	public void init(RpcConfig config) throws Exception {
		disruptor = new Disruptor<RpcUnit<T>>(new ProcessUnitFactory(), 16, Executors.defaultThreadFactory());
		disruptor.handleEventsWith(new ProcessUnitEventHandler(this));
		ringBuffer = disruptor.start();
		this.rpcConfig = config;
	}

	@Override
	public void process(RpcUnit<T> unit) throws Exception {
		rpcConfig.getRpcCallback().callback(unit);
	}

	@Override
	public boolean receive(Session session, T msg) {
		ringBuffer.publishEvent(TRANSLATOR, session, msg);
		return true;
	}

	@Override
	public void shutdown() {
		disruptor.shutdown();
	}

	class ProcessUnitEventHandler implements EventHandler<RpcUnit<T>> {
		private MessageProcessor<T> processor;

		public ProcessUnitEventHandler(MessageProcessor<T> processor) {
			this.processor = processor;
		}

		public void onEvent(RpcUnit<T> unit, long sequence, boolean endOfBatch) {
			try {
				processor.process(unit);
			} catch (Exception e) {
				LOGGER.catching(e);
			}

		}
	}

	class ProcessUnitFactory implements EventFactory<RpcUnit<T>> {
		public RpcUnit<T> newInstance() {
			return new RpcUnit<T>();
		}
	}
}
